package ay.shadow.common.rabbitmq.demo;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * Direct（直连交换机）：根据精确匹配路由键（Routing Key）来将消息转发到绑定了相同路由键的队列。
 * Fanout（扇出交换机）：不处理路由键，直接将消息广播给绑定到该交换机的所有队列。类似广播机制。
 * Topic（主题交换机）：根据路由键的模式匹配（可以使用通配符）将消息路由到符合条件的队列。
 * Headers（头交换机）：根据消息的 headers（头部信息）属性来路由，而不是使用路由键。
 */
public class RbMqDemo {
    private final static String QUEUE_NAME = "seckillQueue";
    private static final String EXCHANGE_NAME = "shadow-exchange";
    private static final String EXCHANGE_DIRECT_NAME = "shadow-exchange-direct";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("8.130.103.38");
        factory.setUsername("admin");
        factory.setPassword("admin");
        try {
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
//            ordinary(channel);
//            pubSub(channel);
            ordinary(channel);
        } catch (Exception e) {

        }
    }

    /**
     * Routing模式
     */
    public static void Routing(Channel channel) throws IOException {
        channel.exchangeDeclare(EXCHANGE_DIRECT_NAME, BuiltinExchangeType.DIRECT.getType());
        //指定的可以
        String severity = "info.*";
        //消息内容
        String message = "directMsg";
        channel.basicPublish(EXCHANGE_DIRECT_NAME, severity, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + severity + "':'" + message + "'");

//        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info.ay");
        channel.queueBind("德国", EXCHANGE_DIRECT_NAME, "error");
        channel.queueBind("德国", EXCHANGE_DIRECT_NAME, "warning");
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message1 = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message1 + "'");
        };
        channel.basicConsume("德国",true, deliverCallback, consumerTag -> {});
    }

    /**
     * Pub/Sub模式
     */
    public static void pubSub(Channel channel) throws IOException {
        //四种交换机类型，分别是Direct 、Fanout 、Topic 、Headers
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT.getType());
        String message = "Hello World!";
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
//        String queueName = channel.queueDeclare().getQueue();
        //将队列和交换机绑定
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 消费需要重开一个线程才能消费消息 只是举例
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message1 = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message1 + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    }

    /**
     * 普通模式
     */
    public static void ordinary(Channel channel) throws IOException {
        //创建
        channel.queueDeclare(QUEUE_NAME, true, true, true, null);
        String message = "Hello World!";
        //发送
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");
//            Channel channel = connection.createChannel();
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message1 = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message1 + "'");
        };
        //channel.basicQos(1);  消费初始信息量
        //消费
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
        //删除 走不到这个删除 举个例
        channel.queueDelete(QUEUE_NAME);
    }
}
